package cn._51doit.flink.day08;

import cn._51doit.flink.day07.KafkaToRedisWordCount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.util.Collector;

import java.util.Properties;

/**
 * HashMapStateBackend
 * 优点：可以存储大量状态、长窗口、单个k-v比较大的状态并且可以配置高可用，可以将数据保存到HDFS中
 *
 * StateBackend有两种方式
 * 1.为每一个单独个性化配置，在代码中要env.setStateBackend(new HashMapStateBackend());
 * 2.为整个集群配置全局的StateBackend，需要在flink的配置文件中添加
 *      state.backend: hashmap
 *      state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints
 */
public class HashMapStateBackendDemo {

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //开启Checkpointing
        env.enableCheckpointing(10000);
        //设置StateBackend为HashMapStateBackend
        env.setStateBackend(new HashMapStateBackend());
        //将checkpoint数据保存到外部的hdfs中
        env.getCheckpointConfig().setCheckpointStorage("hdfs://node-1.51doit.cn:9000/chk001");

        //默认值ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION，当job被cancel后，外部的checkpoint数据会被删除
        //ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION,当job被cancel后，保留外部的checkpoint数据
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //创建KafkaSource
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092");
        //properties.setProperty("group.id", "ggg28");
        //没有偏移量从头开始读，有偏移量接着偏移量读
        properties.setProperty("auto.offset.reset", "earliest");
        //不将偏移量写入到Kafka特殊的topic中
        properties.setProperty("enable.auto.commit", "false");
        //调用addSource添加Flink的Connector
        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<String>(
                "wordcount28",
                new SimpleStringSchema(),
                properties
        );
        //在checkpoint时不将偏移量写入到Kafka特殊的topic
        //将偏移量写入到OperatorState中
        flinkKafkaConsumer.setCommitOffsetsOnCheckpoints(false);

        DataStreamSource<String> lines = env.addSource(flinkKafkaConsumer);

        //将数据进行切分聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = line.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = wordAndOne.keyBy(t -> t.f0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> res = keyedStream.sum(1);

        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()
                .setHost("node-2.51doit.cn")
                .setPassword("123456")
                .setDatabase(4)
                .build();


        //将数据输出到Redis
        res.addSink(new RedisSink<Tuple2<String, Integer>>(conf, new KafkaToRedisWordCount.RedisWordCountMapper()));

        env.execute();

    }

}
